package nthu.web.model.domain;

import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.URI;

import nthu.web.model.domain.box.ABox;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.json.JSONException;
import org.json.JSONObject;
import org.json.XML;

public class Hadoop {
	
	Configuration conf = new Configuration();
	FileSystem fs = null;
	public Hadoop() throws IOException{
		this.fs = FileSystem.get(URI.create("hdfs://localhost:9000/user/bear"),conf);
	}
	
	public Hadoop(URI fsPath) throws IOException{
		this.fs = FileSystem.get(fsPath, conf);
	}
	
	public boolean store(Object input, String fileName, String fileDst) throws IOException{
		
		byte[] inputByte;

		ByteArrayOutputStream bos = new ByteArrayOutputStream();
		try{
			ObjectOutputStream oos = new ObjectOutputStream(bos);
			oos.writeObject(input);
			oos.flush();
			oos.close();
			bos.close();
			inputByte = bos.toByteArray();
		}catch(IOException ex){
			inputByte = null;
		}
		
		if(this.fs != null && inputByte !=null){
			OutputStream out = this.fs.create(new Path(fileDst+"/"+fileName));
			out.write(inputByte);
			return true;
		}
		return false;
	}
	
	public boolean storeMemoBox(ABox box, String fileDst) throws JSONException, IOException{
		JSONObject json = new JSONObject();
		try {
			json.put("id", box.getId());
			json.put("cid", box.getCreateUserId());
			json.put("name", XML.escape(box.getName()));
			json.put("Content", XML.escape(box.getContent())); //bug can't use "content" as key
			json.put("stamp", box.getStamp());
		} catch (JSONException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		OutputStream out = this.fs.create(new Path(fileDst),true);
		DataOutputStream dos = new DataOutputStream(out);
		//dos.writeBytes(XML.toString(json));
		dos.writeBytes(XML.toString(json));
		dos.flush();
		dos.close();
		out.close();
		return true;
	}
	
	@SuppressWarnings("deprecation")
	public JSONObject getMemoBox(String fileDst) throws IOException, JSONException{
		InputStream ins = this.fs.open(new Path(fileDst));
		DataInputStream dis = new DataInputStream(ins);
		String buffer;
		String ans = null;
		while((buffer = dis.readLine())!=null){
			ans +=buffer;
		}
		return XML.toJSONObject(ans);
	}
	
	public ByteArrayOutputStream readByte(String fileDst){
		try{
			InputStream in = fs.open(new Path(fileDst));
			DataInputStream inData = new DataInputStream(in);
			ByteArrayOutputStream buffer = new ByteArrayOutputStream();
			IOUtils.copyBytes(inData, buffer, conf);
			return buffer;
		}catch(IOException ex){
			return null;
		}
	}
}
